package com.zetyun.tiger.connector.cdc;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.connectors.redis.common.container.RedisContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisSentinelPool;

import java.io.Closeable;
import java.io.IOException;
import java.util.Objects;

@Slf4j
public class SimpleRedisContainer implements SimpleRedisCommandsContainer, Closeable {

    private static final long serialVersionUID = 1L;

    private transient JedisSentinelPool jedisSentinelPool;


    /**
     * Use this constructor if Redis environment is clustered with sentinels.
     *
     * @param sentinelPool SentinelPool which actually manages Jedis instances
     */
    public SimpleRedisContainer(final JedisSentinelPool sentinelPool) {
        Objects.requireNonNull(sentinelPool, "Jedis Sentinel Pool can not be null");
        this.jedisSentinelPool = sentinelPool;
    }

    @Override
    public void open() throws Exception {

        // echo() tries to open a connection and echos back the
        // message passed as argument. Here we use it to monitor
        // if we can communicate with the cluster.

        getInstance().echo("Test");
    }

    @Override
    public void hset(String key, String hashField, String value, Integer ttl) {

        Jedis jedis = null;
        try {
            jedis = getInstance();
            jedis.hset(key, hashField, value);
            if (ttl != null) {
                jedis.expire(key, ttl);
            }
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HSET to key {} and hashField {} error message {}",
                        key, hashField, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public String hget(String key, String hashField) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.hget(key, hashField);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HGET to key {} and hashField {} error message {}",
                        key, hashField, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public String get(String key) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.get(key);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command GET to key {}  error message {}",
                        key,  e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }


    @Override
    public Boolean hexists(String key, String hashField) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.hexists(key, hashField);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HEXISTS to key {} and hashField {} error message {}",
                        key, hashField, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long sadd(String key, String... members) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.sadd(key, members);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command SADD to key {} and hashField {} error message {}",
                        key, members, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }


    @Override
    public Long hlen(String key) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.hlen(key);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HLEN to key {}  error message {}",
                        key, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long expire(String key, Integer seconds) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.expire(key,seconds);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command EXPIRE to key {} and seconds {} error message {}",
                        key, seconds, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long hdel(String key, String... fields) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.hdel(key, fields);

        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HDEL to key {} and hashField {} error message {}",
                        key, fields, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long hincrBy(final String key, final String hashField, final Long value, final Integer ttl) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            Long resValue = jedis.hincrBy(key, hashField, value);
            if (ttl != null) {
                jedis.expire(key, ttl);
            }
            return resValue;
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command HINCRBY to key {} and hashField {} error message {}",
                        key, hashField, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public void sadd(String setName, String value, final Integer ttl) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            jedis.sadd(setName, value);
            if (ttl != null) {
                jedis.expire(setName, ttl);
            }
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command SADD to key {} and value {} error message {}",
                        setName, value, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long srem(String key, final Integer ttl, String... members) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            Long resValue = jedis.srem(key, members);
            if (ttl != null) {
                jedis.expire(key, ttl);
            }
            return resValue;
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command SREM to key {} and members {} error message {}",
                        key, members, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }

    @Override
    public Long scard(String key) {
        Jedis jedis = null;
        try {
            jedis = getInstance();
            return jedis.scard(key);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("Cannot send Redis message with command SCARD to key {} error message {}",
                        key, e.getMessage());
            }
            throw e;
        } finally {
            releaseInstance(jedis);
        }
    }


    /**
     * Closes the Jedis instances.
     */
    @Override
    public void close() throws IOException {
        if (this.jedisSentinelPool != null) {
            this.jedisSentinelPool.close();
        }
    }


    /**
     * Returns Jedis instance from the pool.
     *
     * @return the Jedis instance
     */
    private Jedis getInstance() {
        if (jedisSentinelPool != null) {
            return jedisSentinelPool.getResource();
        }
        throw new IllegalStateException("Jedis Sentinel Pool can not be null");
    }


    /**
     * Closes the jedis instance after finishing the command.
     *
     * @param jedis The jedis instance
     */
    private void releaseInstance(final Jedis jedis) {
        if (jedis == null) {
            return;
        }
        try {
            jedis.close();
        } catch (Exception e) {
            log.error("Failed to close (return) instance to pool", e);
        }
    }
}
